package github.sf.fw.utils;


import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Field;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

@Slf4j
public class AsyncHelper {
    private static int corePoolSize = 8;
    private static int maximumPoolSie = 10;
    private static String poolName = "self_async_helper";
    /**
     * 声明future缓存
     */
    private final ConcurrentHashMap<String, Future<?>> futures = new ConcurrentHashMap<>();
    /**
     * 初始化线程池
     */
    private ThreadPoolExecutor pool = null;

    private AsyncHelper() {
        pool = new ThreadPoolExecutor(corePoolSize, maximumPoolSie, 15, TimeUnit.SECONDS, new LinkedBlockingDeque<>(100), new SelfThreadFactory(poolName), new ThreadPoolExecutor.DiscardPolicy());
        //允许设置核心线程超时后的线程池超时自毁
        pool.allowCoreThreadTimeOut(true);
    }

    public static void setCorePoolSize(int corePoolSize) {
        AsyncHelper.corePoolSize = corePoolSize;
    }

    public static void setMaximumPoolSie(int maximumPoolSie) {
        AsyncHelper.maximumPoolSie = maximumPoolSie;
    }

    /**
     * 懒加载，单例模式
     */
    public static AsyncHelper getDefaultHelper() {
        return factory.instance;
    }


    public static AsyncHelper getNewHelper(int corePoolSize, int maximumPoolSie, String poolName) {
        AsyncHelper.setCorePoolSize(corePoolSize);
        AsyncHelper.setMaximumPoolSie(maximumPoolSie);
        AsyncHelper.poolName = poolName;
        return new AsyncHelper();
    }

    private void checkAndRunConsumer(String key, Consumer<String> add) {
        if (futures.containsKey(key)) {
            log.trace("重复key:" + key);
            Future<?> future = futures.get(key);
            FutureTask<?> task = (FutureTask<?>) future;
            try {
                Field field = FutureTask.class.getDeclaredField("outcome");
                field.setAccessible(true);
                Object outcome = field.get(task);
                if (outcome != null) {
                    futures.remove(key);
                    log.error("检测到上一当前id的线程运行出现异常，已将其移除队列");
                    throw new RuntimeException("程序上次处理失败。当前已中断，请重试\n", (Throwable) outcome);
                } else {
                    if (future.isDone()) {
                        futures.remove(key);
                        add.accept(key);
                    } else {
                        log.error("上一key:" + key + "还未执行完成，不会提交该任务");
                        throw new RuntimeException("任务已经提交，请勿重复提交");
                    }
                }
            } catch (IllegalAccessException | NoSuchFieldException e) {
                throw new RuntimeException(e.getMessage());
            }
        } else {
            add.accept(key);
        }
    }

    private void add(Runnable runnable, String key) {
        Future<?> future = pool.submit(runnable);
        futures.putIfAbsent(key, future);
    }

    private <V> void add(Callable<V> callable, String name) {
        Future<?> future = pool.submit(callable);
        futures.putIfAbsent(name, future);
    }

    /**
     * 提交有返回值的任务
     */
    public <V> void async(Callable<V> callable, String name) {
        checkAndRunConsumer(name, (k) -> add(callable, k));
    }

    public void async(Runnable runnable, String name) {
        checkAndRunConsumer(name, (k) -> add(runnable, k));
    }

    /**
     * 根据name获取future
     */
    public Future<?> getFuture(String name) {
        return futures.get(name);
    }

    /**
     * 实现懒加载，线程安全
     */
    private static class factory {
        private static final AsyncHelper instance = new AsyncHelper();
    }

    static class SelfThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        SelfThreadFactory(String poolName) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            if (poolName == null) {
                namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
            } else {
                namePrefix = poolName;
            }

        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (t.isDaemon()) t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }

    }
}